package producer;

import base.MQBase;
import consumer.ConsumerJsonBody;
import consumer.bo.NormalBody;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import serializeUtil.JDKSerialize;

import java.nio.charset.StandardCharsets;
import java.util.Map;

/**
 * @author : zGame
 * @version V1.0
 * @Project: rokectmqprac
 * @Package producer
 * @Description: dal消息队列使用模型demo producer,不同子类 tag不同,并且保证了同一tag的顺序
 * @date Date : 2022年01月10日 17:10
 */


/**
 * 一、RockerMQ消息常见发送状态
 * Broker消息投递状态讲解：
 * FLUSH_DISK_TIMEOUT： 没有在规定时间内完成刷盘 (刷盘策略需要为SYNC_FLUSH 才会出这个错误)
 * FLUSH_SLAVE_TIMEOUT：主从模式下，broker是SYNC_MASTER, 没有在规定时间内完成主从同步
 * SLAVE_NOT_AVAILABLE：从模式下，broker是SYNC_MASTER, 但是没有找到被配置成Slave的Broker
 * SEND_OK：发送成功
 *
 * 二、RocketMQ消息生产和消费异常重试和阈值设定：
 * 消息重投(保证数据的高可靠性),本身内部支持重试，默认次数是2，
 * 如果网络情况比较差，或者跨集群则建改多几次
 *
 * 消费端重试：
 * 原因：消息处理异常、broker端到consumer端各种问题，如网络原因闪断，消费处理失败，ACK返回失败等等问题。
 * 注意：
 * (1)、重试间隔时间配置 ,默认每条消息最多重试 16 次
 * messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
 * (2)、超过重试次数人工补偿
 * (3)、消费端去重：一条消息无论重试多少次，这些重试消息的 Message ID，key 不会改变。
 * (4)、消费重试只针对集群消费方式生效；广播方式不提供失败重试特性，即消费失败后，失败消息不再重试，继续消费新的消息。
 */
public abstract class ProducerUtil extends MQBase {
    
    protected DefaultMQProducer producer;
    
    public ProducerUtil() {
        try {
            producer = new DefaultMQProducer(getGroup());
            propertySet();
            producer.start();
            
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
    
    public void mqSend(int serverId, ConsumerJsonBody body)
            throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        //充当tag arg
        String param = this.getClass().getSimpleName();
        // 在这里把线程参数key,线程tag,处理handler分配好,消费者的msgdispatcher直接负责分发,解包操作放到对应逻辑线程去做
        
        Message msg = new Message(getTopic(serverId),JDKSerialize.convert2Byte(body));
        Map<String, String> prt = msg.getProperties();
        prt.put("threadTag",body.getThreadTag());
        prt.put("threadKey",body.getThreadKey());
        prt.put("handlerKey",body.getClass().getSimpleName());
        //sequence & (arrayLength-1) = arrayIndex,
        producer.send(msg, (v1, v2, v3) -> {
            //mq默认4个queue，可以快速取模
            int index = v3.hashCode() & (v1.size() - 1);
            return v1.get(index);
        }, param);
    }

}
